# uncompyle6 version 3.2.3
# Python bytecode 3.6 (3379)
# Decompiled from: Python 3.6.8 |Anaconda custom (64-bit)| (default, Feb 21 2019, 18:30:04) [MSC v.1916 64 bit (AMD64)]
# Embedded file name: server\routers\api\auth\__init__.py
__author__ = "sanyi"
import hashlib, hmac, logging, time, uuid
from server.http_responses.http_errors import *
from server.routers.api.validator import validate_value

logger = logging.getLogger("swagger.api.auth")


class XAuthFactory:
    def __init__(self, rate_limiter=None):
        self.resolvers = {}
        self.rate_limiter = rate_limiter

    def register_auth_type(self, auth_class):
        self.resolvers[auth_class.TYPE_ID] = auth_class

    def get(self, request, remote_address=None):
        if remote_address is None:
            remote_address = request.environ.get(
                "HTTP_X_REAL_IP",
                request.environ.get(
                    "HTTP_X_FORWARDED_FOR", request.environ["REMOTE_ADDR"]
                ),
            )
        if self.rate_limiter:
            r = self.rate_limiter.wait_time(remote_address)
            if r:
                raise TooManyRequests(r)
        try:
            auth_data = request.headers.get("X-Auth", None)
            if not auth_data:
                logger.debug("x_auth validation failed: empty")
                raise InvalidAuth()
            problems, _ = validate_value(
                auth_data, dict(type="string", format="hex", minimum=33)
            )
            if problems:
                logger.debug("x_auth validation failed: %s", problems)
                raise InvalidAuth()
            try:
                auth_class = self.resolvers[auth_data[0]]
            except KeyError:
                logger.debug("x_auth validation failed: unknown auth %s", auth_data[0])
                raise InvalidAuth()

            result = auth_class.from_request(request)
            result.remote_address = remote_address
            return result
        except Exception:
            if self.rate_limiter:
                self.rate_limiter.add(remote_address)
            raise


class XAuth:
    def __init__(self, key):
        self.key = key
        self.remote_address = None

    @classmethod
    def from_request(cls, request):
        key = request.headers.get("X-Auth", None)
        return cls(key)

    @property
    def has_session_support(self):
        return False


class ApiStatusAuth(XAuth):
    TYPE_ID = "0"

    @classmethod
    def from_request(cls, request):
        key = request.headers.get("X-Auth", None)
        if key is None or len(key) != 33 or key[0] != cls.TYPE_ID:
            print("ApiStatusAuth: invalid length")
            raise InvalidAuth()
        return cls(key)

    @classmethod
    def create(cls, api_key):
        return cls(cls.TYPE_ID + api_key)

    @property
    def api_key(self):
        return self.key[1:]


class ApiScanAuth(ApiStatusAuth):
    pass


class ApiKeyAuth(XAuth):
    TYPE_ID = "1"

    @classmethod
    def from_request(cls, request):
        key = request.headers.get("X-Auth", None)
        if key is None or len(key) != 65 or key[0] != cls.TYPE_ID:
            raise InvalidAuth()
        return cls(key)

    @classmethod
    def create(cls, user_id, api_key):
        return cls(cls.TYPE_ID + user_id.replace("-", "") + api_key.replace("-", ""))

    @property
    def user_id(self):
        return self.key[1:33]

    @property
    def admin_id(self):
        pass

    @property
    def api_key(self):
        return self.key[33:]


class UiSessionAuth(XAuth):
    TYPE_ID = "2"
    SESSION_SECRET = "session-secret-12345678890"

    @classmethod
    def create(cls, user_id, expiration, random_part=None):
        if random_part:
            if not len(random_part) == 24:
                raise AssertionError
            else:
                random_part = uuid.uuid4().hex[0:24]
            data = "%s%08x%s" % (user_id.replace("-", ""), int(expiration), random_part)
            auth_hash = hmac.new(
                cls.SESSION_SECRET.encode(), data.encode(), digestmod=hashlib.sha256
            )
            return cls(cls.TYPE_ID + data + auth_hash.hexdigest())

    @classmethod
    def from_request(cls, request):
        key = request.headers.get("X-Auth", None)
        if not key:
            logger.debug("UiSessionAuth: no key")
            raise InvalidAuth()
        if key[0] != cls.TYPE_ID:
            logger.debug("UiSessionAuth: invalid type")
            raise InvalidAuth()
        if len(key) != 129:
            logger.debug("UiSessionAuth: invalid length")
            raise InvalidAuth()
        if request.cookies.get("ui_session") != key:
            logger.debug("UiSessionAuth: invalid cookie")
            raise InvalidAuth()
        expiration = int(key[33:41], 16)
        if time.time() > expiration:
            raise SessionExpired()
        request_signature = key[65:]
        signature = (
            hmac.new(
                cls.SESSION_SECRET.encode(),
                key[1:65].encode(),
                digestmod=hashlib.sha256,
            )
        ).hexdigest()
        if signature != request_signature:
            logger.debug("UiSessionAuth: invalid hex digest")
            raise InvalidSession()
        return cls(key)

    @property
    def expiration(self):
        return int(self.key[33:41], 16)

    @property
    def user_id(self):
        return self.key[1:33]

    @property
    def admin_id(self):
        pass

    @property
    def has_session_support(self):
        return True


class AdminImpersonatorAuth(XAuth):
    TYPE_ID = "3"

    @classmethod
    def create(cls, user_id, admin_id, admin_auth, expiration, random_part=None):
        if random_part:
            if not len(random_part) == 24:
                raise AssertionError
            else:
                random_part = uuid.uuid4().hex[0:24]
            data = "%s%s%s%08x%s" % (
                user_id.replace("-", ""),
                admin_id.replace("-", ""),
                admin_auth,
                int(expiration),
                random_part,
            )
            auth_hash = hmac.new(
                UiSessionAuth.SESSION_SECRET.encode(),
                data.encode(),
                digestmod=hashlib.sha256,
            )
            return cls(cls.TYPE_ID + data + auth_hash.hexdigest())

    @classmethod
    def from_request(cls, request):
        key = request.headers.get("X-Auth", None)
        if not key:
            logger.debug("UiSessionAuth: no key")
            raise InvalidAuth()
        if key[0] != cls.TYPE_ID:
            logger.debug("UiSessionAuth: invalid type")
            raise InvalidAuth()
        if len(key) != 225:
            logger.debug("UiSessionAuth: invalid length")
            raise InvalidAuth()
        if request.cookies.get("ui_session") != key:
            logger.debug("UiSessionAuth: invalid cookie")
            raise InvalidAuth()
        expiration = int(key[129:137], 16)
        if time.time() > expiration:
            raise SessionExpired()
        request_signature = key[161:]
        signature = (
            hmac.new(
                UiSessionAuth.SESSION_SECRET.encode(),
                key[1:161].encode(),
                digestmod=hashlib.sha256,
            )
        ).hexdigest()
        if signature != request_signature:
            logger.debug("hex digest failed")
            raise InvalidSession()
        return cls(key)

    @property
    def user_id(self):
        return self.key[1:33]

    @property
    def admin_id(self):
        return self.key[33:65]

    @property
    def expiration(self):
        return int(self.key[129:137], 16)

    @property
    def admin_auth(self):
        return self.key[65:129]

    @property
    def has_session_support(self):
        return True
